package com.tcl.kafka;

import com.alibaba.cloud.nacos.NacosConfigManager;
import com.tcl.kafka.config.KafkaStartPauseProperties;
import com.tcl.kafka.constant.StartPauseConstant;
import com.tcl.kafka.core.KafkaService;
import com.tcl.kafka.listener.NacosConfigListener;
import com.tcl.kafka.scheduled.StartPauseScheduled;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.scheduling.annotation.EnableScheduling;


/**
 * TclKafkaAutoConfiguration
 * 自动注入配置类
 *
 * @author liangxi.zeng@tcl.com
 */
@Configuration
@EnableConfigurationProperties(KafkaStartPauseProperties.class)
@ConditionalOnProperty(value = "tcl.kafka.enabled",havingValue="true",matchIfMissing=true)
@Slf4j
@EnableScheduling
public class TclKafkaAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public KafkaService kafkaService(KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry
            ,KafkaStartPauseProperties kafkaStartPauseProperties) {
        return new KafkaService(kafkaListenerEndpointRegistry,kafkaStartPauseProperties);
    }

    @Bean
    @ConditionalOnProperty(name = StartPauseConstant.START_PAUSE_TYPE, havingValue = StartPauseConstant.START_PAUSE_TYPE_CONFIG)
    public NacosConfigListener nacosConfigListener(NacosConfigManager nacosConfigManager,KafkaService kafkaService
            ,KafkaStartPauseProperties kafkaStartPauseProperties) {
        log.info("启停类型nacos配置监听");
        return new NacosConfigListener(nacosConfigManager,kafkaService,kafkaStartPauseProperties);
    }

    @Bean
    @ConditionalOnProperty(name = StartPauseConstant.START_PAUSE_TYPE, havingValue = StartPauseConstant.START_PAUSE_TYPE_SCHEDULED)
    public StartPauseScheduled StartPauseScheduled(KafkaService kafkaService
            , KafkaStartPauseProperties kafkaStartPauseProperties) {
        log.info("启停类型定时任务");
        return new StartPauseScheduled(kafkaService,kafkaStartPauseProperties);
    }




}
